package org.example.common.configs.queue.disruptor;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.lang.Assert;
import com.google.common.base.Throwables;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ExceptionHandlerWrapper;
import com.lmax.disruptor.dsl.ProducerType;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.PayloadApplicationEvent;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.context.event.SimpleApplicationEventMulticaster;
import org.springframework.context.support.AbstractApplicationContext;

import java.util.stream.Stream;

/**
 * @author weigang
 * @create 2022-07-11 14:16
 */
@Configuration
@EnableConfigurationProperties(DisruptorConfig.class)
@AllArgsConstructor
@Slf4j
// 介入Spring的启动过程。在Spring容器将所有的Bean都初始化完成之后，做一些操作，这个时候我们就可以实现一个接口 implements ApplicationListener<ContextRefreshedEvent>
public class DisruptorConfigurer implements ApplicationListener<ContextRefreshedEvent> {

    // Spring应用上下文中最重要的一个类，这个抽象类中提供了几乎ApplicationContext的所有操作。
    // 主要有容器工厂的处理，事件的发送广播，监听器添加，容器初始化操作refresh方法，然后就是bean的生成获取方法接口等。
    private AbstractApplicationContext applicationContext;
    private ConfigurableBeanFactory beanFactory;
    private DisruptorConfig disruptorConfig;
    private DisruptorEventProducer disruptorEventProducer;
    private DisruptorStat disruptorStat;


    @Override
    // 需要执行的逻辑代码，onApplicationEvent 当spring容器初始化完成后就会执行该方法
    // 存在执行onApplicationEvent两次的问题，详情百度
    public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {

        // 初始化Disruptor Producer，因为Spring中@EventListener创建Bean的时机在 SmartInitializingSingleton#afterSingletonsInstantiated之后
        // 所以只能等上下文完成刷新后，才能获取到所有的@EventListener对应的Bean ==> 见 DefaultEventListenerFactory、EventListenerMethodProcessor
        // 所以注意避免为Spring上下文广播配置异步消费事件

        // 这里使用没有线程池的广播器，会立即执行事件，从而只起到事件路由的功能
        // 观察者模式
        // SimpleApplicationEventMulticaster 是 AbstractApplicationContext 的默认事件广播器
        SimpleApplicationEventMulticaster multicaster = new SimpleApplicationEventMulticaster();
        multicaster.setBeanFactory(beanFactory);
        multicaster.setTaskExecutor(null); //没有使用线程池
        multicaster.setErrorHandler(t ->
                log.warn("Disruptor Multicaster Event Failed, err:{}", Throwables.getStackTraceAsString(t)));
        // 添加 @EventListener Bean到广播器
        if (CollUtil.isNotEmpty(applicationContext.getApplicationListeners())) {
            applicationContext.getApplicationListeners().forEach(multicaster::addApplicationListener);
        }

        // 构造事件分发Disruptor
        Disruptor<DisruptorEvent> disruptor = new Disruptor<>(
                new DisruptorEventFactory(), // 通过事件工厂EventFactory在RingBuffer中预创建事件Event的实例
                disruptorConfig.getBufferSize(), // buffer size 环形队列大小，2的指数
                new ThreadFactoryBuilder()  // 消费者线程池
                        .setNameFormat("disruptor-thread-%d")
                        .setUncaughtExceptionHandler((thread, ex) ->
                                log.info("disruptor pool raised error, ex:{}", Throwables.getStackTraceAsString(ex)))
                        .build(),
                // Single模式的生产者，似乎不是线程安全的
                // MULTI模式，使用CompareAndSet模式，应该是支持多线程并发访问
                // 多生产者模式
                ProducerType.MULTI,
                // 消费者等待环形队列的指定位置有可用数据时的阻塞策略，共同消费模式好像是用不到
                // 生产者好像只是用这个来通知消费者，有可用消息可用消费了
                // 等待策略，决定消费者如何等待生产者
                new BlockingWaitStrategy());


        Assert.isTrue(disruptorConfig.getConsumerCnt() > 0);
        // 根据消费者数量，创建多线程消费者
        DisruptorEventHandler[] consumers = Stream.iterate(0, t -> t)
                .limit(disruptorConfig.getConsumerCnt())
                .map(t -> new DisruptorEventHandler(event -> {
                    Assert.isFalse(event instanceof ApplicationEvent);
                    PayloadApplicationEvent<?> applicationEvent = new PayloadApplicationEvent<>(applicationContext, event);
                    multicaster.multicastEvent(
                            applicationEvent,
                            applicationEvent.getResolvableType());
                }))
                .toArray(DisruptorEventHandler[]::new);

        // Disruptor事件分发器连接消费者处理器DisruptorEventHandler
        // 消费者处理器DisruptorEventHandler实现WorkHandler，是共同消费模式。
        // 共同消费模式 : Disruptor队列中同一个事件可以有多个消费者,消费者之间既可以并行处理,也可以形成依赖图相互依赖,按照先后次序进行处理
        disruptor.handleEventsWithWorkerPool(consumers);

        // 消费异常处理
        ExceptionHandlerWrapper<DisruptorEvent> exceptionHandlerWrapper = new ExceptionHandlerWrapper<>();
        exceptionHandlerWrapper.switchTo(new DisruptorWorkerExceptionHandler());
        disruptor.setDefaultExceptionHandler(exceptionHandlerWrapper);

        //获取RingBuffer环，用于接取生产者生产的事件
        RingBuffer<DisruptorEvent> ringBuffer = disruptor.getRingBuffer();
        DisruptorEventProducerWorker worker = new DisruptorEventProducerWorker(ringBuffer, disruptor, disruptorStat);
        worker.start();
        disruptorEventProducer.setWorker(worker);
    }
}
